package com.swak.reactivex.transport.udp;

import java.net.InetSocketAddress;

import com.swak.reactivex.transport.Message;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.channel.socket.DatagramPacket;
import io.netty.handler.codec.EncoderException;
import io.netty.util.ReferenceCountUtil;

public abstract class MessageToDatagramEncoder extends ChannelOutboundHandlerAdapter {

	protected boolean acceptOutboundMessage(Object msg) throws Exception {
		return msg != null && msg instanceof Message;
	}

	protected abstract void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception;

	@Override
	public void write(ChannelHandlerContext ctx, Object message, ChannelPromise promise) throws Exception {
		if (this.acceptOutboundMessage(message)) {
			ByteBuf buffer = null;
			try {
				buffer = ctx.alloc().ioBuffer();
				Message cast = (Message) message;
				try {
					this.encode(ctx, cast, buffer);
				} finally {
					ReferenceCountUtil.release(message);
				}
				InetSocketAddress address = cast.getAddress();
				ByteBuf sendBuffer = Unpooled.EMPTY_BUFFER;
				if (buffer.isReadable()) {
					sendBuffer = buffer;
				} else {
					buffer.release();
				}
				DatagramPacket packet = new DatagramPacket(sendBuffer, address);
				ctx.write(packet, promise);
				buffer = null;
			} catch (EncoderException e) {
				throw e;
			} catch (Throwable e) {
				throw new EncoderException(e);
			} finally {
				if (buffer != null) {
					buffer.release();
				}
			}
		} else {
			ctx.write(message, promise);
		}
	}
}
